package com.webchat.netty.server.config;

import com.webchat.common.util.JsonUtil;
import com.webchat.netty.server.common.NettyInstanceUtils;
import com.webchat.netty.server.config.properties.RocketMQConfig;
import com.webchat.netty.server.enums.TopicNameEnum;
import com.webchat.netty.server.service.IMMessageHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.List;


@Slf4j
public class RocketMQResource {

    private static DefaultMQProducer producer;

    public static DefaultMQProducer getProducer() {
        return producer;
    }

    private static IMMessageHandler messageHandler = new IMMessageHandler();

    /**
     * 初始化producer配置
     *
     * @param rocketMQConfig
     * @throws Exception
     */
    public static void initProducer(RocketMQConfig rocketMQConfig) throws Exception {

        producer = new DefaultMQProducer(rocketMQConfig.getProducerGroup());
        producer.setNamesrvAddr(rocketMQConfig.getNamesrvAddr());
        producer.setSendMsgTimeout(10000);
        producer.setDefaultTopicQueueNums(8);
//        producer.setSendMsgTimeout(rocketMQConfig.getProducerSendMsgTimeout());
        // 设置发送失败时的重试次数
        producer.setRetryTimesWhenSendFailed(0);
        producer.setRetryAnotherBrokerWhenNotStoreOK(false);
        producer.setSendLatencyFaultEnable(true);
        producer.start();
        addJVMShutdownHook();
    }

    /**
     * 初始化consumer配置
     *
     * @param rocketMQConfig
     * @throws Exception
     */
    public static void initConsumer(RocketMQConfig rocketMQConfig) throws Exception {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(rocketMQConfig.getConsumerGroup());
        consumer.setNamesrvAddr(rocketMQConfig.getNamesrvAddr());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setConsumeThreadMin(rocketMQConfig.getConsumeThreadMin());
        consumer.setConsumeThreadMax(rocketMQConfig.getConsumeThreadMax());
        consumer.subscribe("webchat_ugc_message", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                for (MessageExt msg : list) {
                    log.info("Rocket MQ Consumer Message:{}", JsonUtil.toJsonString(msg));
                    // 收到推送任务，基于Channel管道实现消息推送
                    messageHandler.handleMessage(msg);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        addJVMShutdownHook();
    }

    public static void initIMMessageSendConsumer(RocketMQConfig rocketMQConfig) throws Exception {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(rocketMQConfig.getNodeConsumerGroup());
        consumer.setNamesrvAddr(rocketMQConfig.getNamesrvAddr());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setConsumeThreadMin(rocketMQConfig.getConsumeThreadMin());
        consumer.setConsumeThreadMax(rocketMQConfig.getConsumeThreadMax());
        String instanceMd = NettyInstanceUtils.getInstanceMd5();
        consumer.subscribe(TopicNameEnum.TOPIC_NODE_IM_SEND_MSG.getTopic(), instanceMd);
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                for (MessageExt msg : list) {
                    log.info("Rocket MQ Consumer Message:{}", JsonUtil.toJsonString(msg));
                    // 消息推送
                    messageHandler.doWriteMessage2Receiver(msg);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        addJVMShutdownHook();
    }

    private static void addJVMShutdownHook() {
        // 当前JVM进程挂掉后，优雅的关闭producer 线程
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            destroyProducer();
        }));
    }

    private static void destroyProducer() {
        if (producer != null) {
            producer.shutdown();
        }
    }
}
